package com.gitee.ywj1352.reactor.subcriber;

import org.junit.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.BufferOverflowStrategy;
import reactor.core.publisher.Flux;
import reactor.core.publisher.UnicastProcessor;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.concurrent.CompletableFuture;

public class Demo {

    @Test
    public void fun1(){
        Flux.range(0,10).publishOn(Schedulers.elastic()).subscribe(new BaseSubscriber<Integer>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                subscription.request(1);
            }

            @Override
            protected void hookOnNext(Integer value) {

                System.out.println(Thread.currentThread().getName() + " :vlaue :" + value);
                request(1);
            }

            @Override
            protected void hookOnComplete() {
                super.hookOnComplete();
            }
        });
    }


    @Test
    public void fun2(){
        Flux.range(0,10).subscribe(new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription subscription) {
                subscription.request(1);

            }

            @Override
            public void onNext(Integer integer) {


                System.out.println("value :" + integer);

            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onComplete() {

            }
        });

    }


    /**
     * 抛异常背压
     */
    @Test
    public void backPressure(){
        UnicastProcessor<Integer> hotSource = UnicastProcessor.create();
        Flux<Integer> integerFlux = hotSource
                .publish()
                .autoConnect()
                .onBackpressureError();
        CompletableFuture.runAsync(()->{
            for (int i = 0; i < 100; i++) {
                hotSource.onNext(i);
            }
        });
        integerFlux.subscribe(new BaseSubscriber<Integer>() {

            //初始化拉去
            //这里 的 初始化 拉取的时候  当 你生产 超出 初始化 的时候就会包异常
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                request(10);
            }
            @Override
            protected void hookOnNext(Integer value) {
                super.hookOnNext(value);

                System.out.println("value" + value);
                //当你没拉的时候说完 我这边压力 有点 大 给 生成者报异常
                //request(1);
            }
            @Override
            protected void hookOnComplete() {
                super.hookOnComplete();
            }
        });
    }

    /**
     * 展示 背压访问
     * @throws InterruptedException
     */
    @Test
    public void backPressure2() throws InterruptedException {
        UnicastProcessor<Integer> hotSource = UnicastProcessor.create();
        Flux<Integer> integerFlux = hotSource
                .publish()
                .autoConnect()
                .onBackpressureBuffer(1, BufferOverflowStrategy.ERROR);
        CompletableFuture.runAsync(()->{
            for (int i = 0; i < 11; i++) {
                hotSource.onNext(i);
            }
        });
        Thread.sleep(100);
        BaseSubscriber<Integer> baseSubscriber = new BaseSubscriber<Integer>() {
            //初始化拉去
            //这里 的 初始化 拉取的时候  当 你生产 超出 初始化 的时候就会包异常
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                request(10);
            }

            @Override
            protected void hookOnNext(Integer value) {
                //super.hookOnNext(value);

                System.out.println("value" + value);
                //当你没拉的时候说完 我这边压力 有点 大 给 生成者报异常
                //request(1);
            }
            @Override
            protected void hookOnComplete() {
                super.hookOnComplete();
            }
        };
        integerFlux.subscribe(baseSubscriber);
        baseSubscriber.request(10);
        Thread.sleep(500);
    }



    @Test
    public void test(){

        Flux.range(0,10).subscribe(new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription subscription) {
                subscription.request(1);

            }

            @Override
            public void onNext(Integer integer) {


                System.out.println(Thread.currentThread().getName() + ": value :" + integer);

            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onComplete() {

            }
        });
    }


    @Test
    public void test2() throws InterruptedException {
        Flux.just("tom")
                .map(s -> {
                    System.out.println("(concat @qq.com) at [" + Thread.currentThread() + "]");
                    return s.concat("@qq.com");
                })
                .publishOn(Schedulers.newSingle("thread-a"))
                .map(s -> {
                    System.out.println("(concat foo) at [" + Thread.currentThread() + "]");
                    return s.concat("foo");
                })
                .filter(s -> {
                    System.out.println("(startsWith f) at [" + Thread.currentThread() + "]");
                    return s.startsWith("t");
                })
                .publishOn(Schedulers.newSingle("thread-b"))
                .map(s -> {
                    System.out.println("(to length) at [" + Thread.currentThread() + "]");
                    return s.length();
                })
                .subscribeOn(Schedulers.newSingle("source"))
                .subscribe(System.out::println);

        Thread.sleep(100);
    }


}
